package com.atguigu.gmall.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.JdbcUtil;
import com.atguigu.gmall.realtime.util.RedisUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/8/25 11:45
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private Connection conn;
    private ValueState<Boolean> tableCreatedState;
    private Jedis redis;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        conn = JdbcUtil.getPhoenixConnection(Constant.PHOENIX_URL);
        
        redis = RedisUtil.getRedisClient();
        redis.select(10);
        
        
        tableCreatedState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("tableCreatedState", Boolean.class));
        
        
        
    }
    
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value,
                       Context ctx) throws Exception {
        // 1. 检测需要的表: 如果不存在就创建, 存在就忽略
        checkTable(value);
        // 2. 写入数据
        write2Hbase(value);
        
        // 3. 更新redis缓存
        updateCache(value);
        
    }
    
    private void updateCache(Tuple2<JSONObject, TableProcess> value) {
        String key = value.f1.getSink_table() + ":" + value.f0.getLong("id");
        // 优雅 如果这次是更新, 并且这个维度也在缓存中
        if (value.f1.getOperate_type().equals("update") && redis.exists(key)) {
    
            JSONObject result = new JSONObject();
    
            for (Map.Entry<String, Object> entry : value.f0.entrySet()) {
                result.put(entry.getKey().toUpperCase(), entry.getValue()); // 为了与Phoenix中存储的列名保持大写一致
            }
            redis.set(key, result.toJSONString());
        }
        // 粗暴
        //redis.del(key);
        
    }
    
    private void write2Hbase(Tuple2<JSONObject, TableProcess> value) throws SQLException {
        JSONObject obj = value.f0;
        TableProcess tp = value.f1;
        // 每来一条语句, 执行一次插入
        // upsert into t(id, name, age) values(?,?,?)
        StringBuilder sql = new StringBuilder();
        sql
            .append("upsert into ")
            .append(tp.getSink_table())
            .append("(")
            .append(tp.getSink_columns())
            .append(") values(")
            .append(tp.getSink_columns().replaceAll("[^,]+", "?"))
            .append(")");
        
        PreparedStatement ps = conn.prepareStatement(sql.toString());
        
        String[] columns = tp.getSink_columns().split(",");  // [name, age,...]
        // 给占位符赋值
        for (int i = 0, len = columns.length; i < len; i++) {
            String columnName = columns[i];
            // 1 -> "1"   null -> "null"
            String v = obj.get(columnName) == null ? null : obj.get(columnName).toString();
            ps.setString(i + 1, v);
        }
        
        ps.execute();
        conn.commit();
        ps.close();
        
    }
    
    private void checkTable(Tuple2<JSONObject, TableProcess> value) throws SQLException, IOException {
        if (tableCreatedState.value() == null) {
            // 执行建表语句
            // create table if not exists t(id varchar, name varchar, constraint pk primary key(id) ) SALT_BUCKETS = 4 ;
            StringBuilder sql = new StringBuilder();
            // TODO 拼接建表sql语句
            sql
                .append("create table if not exists ")
                .append(value.f1.getSink_table())
                .append("(")
                .append(value.f1.getSink_columns().replaceAll(",", " varchar,"))
                .append(" varchar, constraint pk primary key(")
                .append(value.f1.getSink_pk() == null ? "id" : value.f1.getSink_pk())
                .append("))")
                .append(value.f1.getSink_extend() == null ? "" : value.f1.getSink_extend());
            
            System.out.println("建表语句: " + sql.toString());
            PreparedStatement ps = conn.prepareStatement(sql.toString());
            ps.execute();
            conn.commit();
            ps.close();
            
            // 更新状态
            tableCreatedState.update(true);
        }
        
    }
    
    @Override
    public void close() throws Exception {
        if (conn != null && !conn.isClosed()) {
            conn.close();
        }
    
        if (redis != null) {
            redis.close(); // 使用连接池获取的客户端, 则close只是把客户端归还给连接池
        }
    }
}
